package cn.com.cennavi.sparkStreamService;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.api.java.function.VoidFunction2;
import org.apache.spark.streaming.Time;

import scala.Tuple2;
import cn.com.cennavi.bean.IndexInputParmBean;

/**
 * 发送平均速度到kafka
 * 
 * @Description
 * @author buf
 * @date 2017年3月9日 下午3:06:36
 */
public class IndexSendAverageSpeedTokafka implements VoidFunction2<JavaPairRDD<Long, IndexInputParmBean>, Time> {
	
	/** @Fields serialVersionUID: */
	  	
	private static final long serialVersionUID = 1L;
	private String metadata_broker_list = null;
	private String outTopic = null;

	public IndexSendAverageSpeedTokafka(String metadata_broker_list1, String outTopic1) {
		this.metadata_broker_list = metadata_broker_list1;
		this.outTopic = outTopic1;
	}

	public void call(JavaPairRDD<Long, IndexInputParmBean> arg0, Time arg1) throws Exception {
		// TODO Auto-generated method stub
		final String timeNow = getHourMin(arg1);
		if (arg0 != null) {
			arg0.foreachPartition(new VoidFunction<Iterator<Tuple2<Long, IndexInputParmBean>>>() {

				
				/** @Fields serialVersionUID: */
				  	
				private static final long serialVersionUID = 1L;

				public void call(Iterator<Tuple2<Long, IndexInputParmBean>> arg0) throws Exception {
					// TODO Auto-generated method stub
					final Properties props = new Properties();
					props.put("metadata.broker.list", metadata_broker_list);
					props.put("serializer.class", "kafka.serializer.StringEncoder");
					props.put("key.serializer.class", "kafka.serializer.StringEncoder");
					props.put("request.required.acks", "-1");
					ProducerConfig config = new ProducerConfig(props);
					// 创建producer
					Producer<String, String> producer = new Producer<String, String>(config);
					List<KeyedMessage<String, String>> datas = new ArrayList<KeyedMessage<String, String>>();
					try {
						while (arg0.hasNext()) {
							Tuple2<Long, IndexInputParmBean> t = arg0.next();
							String linkId = t._1 + "";
							IndexInputParmBean iip = t._2;
							String outLine = iip.toString() + "," + timeNow;
							System.out.println("outLine:"+outLine);
							KeyedMessage<String, String> data = new KeyedMessage<String, String>(outTopic, linkId, outLine);
							datas.add(data);
						}
						producer.send(datas);
						datas.clear();
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						producer.close();
					}
				}
			});
		}
	}

	private SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmm");

	private String getHourMin(Time time) {
		// TODO Auto-generated method stub
		java.util.Date d = new Date();
		d.setTime(time.milliseconds());
		String dateTime = sdf.format(d);
		return dateTime;
	}

}
